package beautiful.butterfly.distributed_task_processing_engine.task_manage.task_data_items_processor.task_data_item_data_manager;

import beautiful.butterfly.distributed_task_processing_engine.task_manage.task_data_items_processor.TaskProcessorComparator;
import beautiful.butterfly.distributed_task_processing_engine.task_manage.task_processor.Statistics;
import beautiful.butterfly.distributed_task_processing_engine.task_manage.task_processor.TaskProcessorInfo;
import beautiful.butterfly.distributed_task_processing_engine.tools.Allots;
import beautiful.butterfly.distributed_task_processing_engine.tools.Constants;
import beautiful.butterfly.distributed_task_processing_engine.tools.TimestampJsonSerializerAndJsonDeserializer;
import beautiful.butterfly.distributed_task_processing_engine.tools.Uuids;
import beautiful.butterfly.distributed_task_processing_engine.user_definition.TaskUserDefinition;
import beautiful.butterfly.distributed_task_processing_engine.zookeeper.ZooKeeperManager;
import beautiful.butterfly.distributed_task_processing_engine.zookeeper.ZooKeepers;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import lombok.Getter;
import lombok.NonNull;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

import java.sql.Timestamp;
import java.util.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;


@Slf4j
public class TaskAndTaskProcessorAndTaskDataItemManage implements ITaskAndTaskProcessorAndTaskDataItemManage
{
    @Setter
    @Getter
    private Gson gson;
    @Setter
    @Getter
    private ZooKeeperManager zooKeeperManager;
    @Setter
    @Getter
    private String tasksRootPath;

    @Setter
    @Getter
    private long zooKeeperBaseDateTimeLong = 0;
    @Setter
    @Getter
    private long localBaseDateTimeLong = 0;

    public TaskAndTaskProcessorAndTaskDataItemManage(ZooKeeperManager zooKeeperManager) throws Exception
    {
        log.warn("初始化" + TaskAndTaskProcessorAndTaskDataItemManage.class.getSimpleName());
        //
        this.gson = new GsonBuilder().registerTypeAdapter(Timestamp.class, new TimestampJsonSerializerAndJsonDeserializer()).setDateFormat("yyyy-MM-dd HH:mm:ss").create();
        //
        this.zooKeeperManager = zooKeeperManager;
        log.warn("创建tasksRootPath节点");
        this.tasksRootPath = this.zooKeeperManager.getRootPath() + Constants.$task_ids;
        if (this.getZooKeeper().exists(this.tasksRootPath, false) == null)
        {
            ZooKeepers.createPath(getZooKeeper(), this.tasksRootPath, this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
        }

        //
        log.warn("校对本地数据和服务器端的时间差");
        String systemTimePath = this.zooKeeperManager.getZooKeeper().create(this.zooKeeperManager.getRootPath() + Constants.$system_time, null, this.zooKeeperManager.getAclList(), CreateMode.EPHEMERAL_SEQUENTIAL);//瞬时节点
        Stat stat = this.zooKeeperManager.getZooKeeper().exists(systemTimePath, false);
        this.zooKeeperBaseDateTimeLong = stat.getCtime();
        //
        ZooKeepers.directoryStructureDelete(getZooKeeper(), systemTimePath);
        //
        this.localBaseDateTimeLong = System.currentTimeMillis();
        if (Math.abs(this.zooKeeperBaseDateTimeLong - this.localBaseDateTimeLong) > 5000)
        {
            log.error("zookeeper " + Math.abs(this.zooKeeperBaseDateTimeLong - this.localBaseDateTimeLong) + " ms");
        }
    }

    public ZooKeeper getZooKeeper() throws Exception
    {
        return this.zooKeeperManager.getZooKeeper();
    }

    public long zooKeeperCurrentTimeMillis()
    {
        return (this.zooKeeperBaseDateTimeLong - this.localBaseDateTimeLong) + System.currentTimeMillis();//求的是zk时间
    }

    /**
     * <pre>
     * ========================================================================================================================
     * ScheduleTask
     * ========================================================================================================================
     * </pre>
     */
    //
    public void createTaskUserDefinition(TaskUserDefinition taskUserDefinition) throws Exception
    {
        String taskIdPath = this.tasksRootPath + Constants.slash + taskUserDefinition.getTaskId();
        String value = this.gson.toJson(taskUserDefinition);
        if (this.getZooKeeper().exists(taskIdPath, false) == null)
        {
            this.getZooKeeper().create(taskIdPath, value.getBytes(), this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
        } else
        {
            throw new IllegalArgumentException("taskId已经存在:" + taskUserDefinition.getTaskId());
        }
    }

    public void updateTaskUserDefinition(TaskUserDefinition taskUserDefinition)
            throws Exception
    {

        String taskIdPath = this.tasksRootPath + Constants.slash + taskUserDefinition.getTaskId();
        String value = this.gson.toJson(taskUserDefinition);
        if (this.getZooKeeper().exists(taskIdPath, false) == null)
        {
            this.getZooKeeper().create(taskIdPath, value.getBytes(), this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
        } else
        {
            this.getZooKeeper().setData(taskIdPath, value.getBytes(), -1);
        }

    }

    @Override
    public void pauseTaskUserDefinition(String taskId) throws Exception
    {
        TaskUserDefinition taskUserDefinition = this.getTaskUserDefinition(taskId);
        taskUserDefinition.setStatus(TaskUserDefinition.status_pause);
        this.updateTaskUserDefinition(taskUserDefinition);
    }


    @Override
    public void resumeTaskUserDefinition(String taskId) throws Exception
    {
        TaskUserDefinition taskUserDefinition = this.getTaskUserDefinition(taskId);
        taskUserDefinition.setStatus(TaskUserDefinition.status_resume);
        this.updateTaskUserDefinition(taskUserDefinition);
    }

    public long setTaskNeedReload(String taskId) throws Exception
    {

        String taskProcessorsPath = this.tasksRootPath + Constants.slash + taskId + Constants.slash + Constants.taskProcessorIds;
        Stat stat = this.getZooKeeper().setData(taskProcessorsPath, "reload=true".getBytes(), -1);
        return stat.getVersion();

    }

    /**
     * zooKeeper清理任务项和处理器,系统会重新分配
     */
    @Override
    public void clearTask(String taskId) throws Exception
    {
        //清除任务项不会报错,清除处理器会报错,ERROR:KeeperErrorCode = Directory not empty for /drds6/tasks/DemoTaskBeanMulti/task_processors
        //如果出现报错,请再点点击进行清理.这样才会把数据清理干净.
        String taskIdPath = this.tasksRootPath + Constants.slash + taskId;
        List<String> list = this.getZooKeeper().getChildren(taskIdPath, false);
        for (String string : list)
        {
            ZooKeepers.directoryStructureDelete(this.getZooKeeper(), taskIdPath + Constants.slash + string);
        }
    }

    @Override
    public void deleteTask(String taskId) throws Exception
    {
        ZooKeepers.directoryStructureDelete(this.getZooKeeper(), this.tasksRootPath + Constants.slash + taskId);
    }

    /**
     * 清除数据后会自动重新分配
     */
    public void clearExpireTask(String taskId, double expireDateInternal) throws Exception
    {
        String taskItemsPath = this.tasksRootPath + Constants.slash + taskId + Constants.slash + Constants.taskItems;
        Stat stat = this.getZooKeeper().exists(taskItemsPath, false);//下面所有的节点均有效
        if (stat != null && zooKeeperCurrentTimeMillis() - stat.getMtime() > (long) (expireDateInternal * 24 * 3600 * 1000))
        {
            //定义不能删除,只能删除处理器信息和任务项
            ZooKeepers.directoryStructureDelete(this.getZooKeeper(), this.tasksRootPath + Constants.slash + taskId + Constants.taskItems);
            ZooKeepers.directoryStructureDelete(this.getZooKeeper(), this.tasksRootPath + Constants.slash + taskId + Constants.taskProcessorIds);
        }
    }

    @Override
    public List<TaskUserDefinition> getTaskUserDefinitionList() throws Exception
    {
        List<TaskUserDefinition> taskUserDefinitionList = new ArrayList<TaskUserDefinition>();
        List<String> taskIdList = this.getZooKeeper().getChildren(this.tasksRootPath, false);
        Collections.sort(taskIdList);
        for (String taskId : taskIdList)
        {
            TaskUserDefinition taskUserDefinition = this.getTaskUserDefinition(taskId);
            if (taskUserDefinition == null)
            {
                throw new NullPointerException(TaskUserDefinition.class.getSimpleName());
            }
            taskUserDefinitionList.add(taskUserDefinition);
        }
        return taskUserDefinitionList;
    }

    public TaskUserDefinition getTaskUserDefinition(String taskId)
            throws Exception
    {
        String taskIdPath = this.tasksRootPath + Constants.slash + taskId;
        if (this.getZooKeeper().exists(taskIdPath, false) == null)
        {
            return null;
        }
        String value = new String(this.getZooKeeper().getData(taskIdPath, false, null));
        TaskUserDefinition taskUserDefinition = this.gson.fromJson(value, TaskUserDefinition.class);
        taskUserDefinition.setTaskId(taskId);//修复taskId属性之前命名错误的问题.
        return taskUserDefinition;
    }

    /**
     * <pre>
     * ========================================================================================================================
     * TaskProcessorInfo
     * ========================================================================================================================
     * </pre>
     */
    /**
     * 会生成新的处理器id
     */
    public void registerTaskProcessorInfoWithSerialNumberAndSetRegisterTrue(@NonNull TaskProcessorInfo taskProcessorInfo) throws Exception
    {
        if (taskProcessorInfo.isRegister())
        {
            throw new Exception(taskProcessorInfo.getTaskProcessorId() + " ");
        }
        //
        String taskIdPath = this.tasksRootPath + Constants.slash + taskProcessorInfo.getTaskId();
        if (this.getZooKeeper().exists(taskIdPath, false) == null)
        {
            this.getZooKeeper().create(taskIdPath, null, this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
        }
        //
        String taskProcessorsPath = taskIdPath + Constants.slash + Constants.taskProcessorIds;
        if (this.getZooKeeper().exists(taskProcessorsPath, false) == null)
        {
            this.getZooKeeper().create(taskProcessorsPath, null, this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
        }
        //
        String uuid = Uuids.getUuid();
        String taskProcessorPath = taskProcessorsPath + Constants.slash + taskProcessorInfo.getTaskId() + Constants.$ + taskProcessorInfo.getIp() + Constants.$ + uuid + Constants.$;
        String path = this.getZooKeeper().create(taskProcessorPath, null, this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT_SEQUENTIAL);
        taskProcessorPath = path;
        //
        path = path.substring(path.lastIndexOf("/") + 1);
        taskProcessorInfo.setTaskProcessorId(path);
        //
        Timestamp heartBeatTime = new Timestamp(this.zooKeeperCurrentTimeMillis());
        taskProcessorInfo.setHeartBeatTimestamp(heartBeatTime);
        //
        String value = this.gson.toJson(taskProcessorInfo);
        this.getZooKeeper().setData(taskProcessorPath, value.getBytes(), -1);//taskProcessor注册在serverPath
        taskProcessorInfo.setRegister(true);
    }

    public void deleteTaskProcessor(String taskId, String taskProcessorId) throws Exception//delete path
    {

        String taskProcessorIdPath = this.tasksRootPath + Constants.slash + taskId + Constants.slash + Constants.taskProcessorIds + Constants.slash + taskProcessorId;
        if (this.getZooKeeper().exists(taskProcessorIdPath, false) != null)
        {
            this.getZooKeeper().delete(taskProcessorIdPath, -1);
        }
    }

    public boolean setRegisterFalseIfNodeIsNotExistedOtherwiseUpdateHeartBeatInfo(TaskProcessorInfo taskProcessorInfo) throws Exception
    {
        Timestamp heartBeatTimestamp = new Timestamp(this.zooKeeperCurrentTimeMillis());
        String taskProcessorIdPath =//
                this.tasksRootPath +//
                        Constants.slash + taskProcessorInfo.getTaskId()//
                        + Constants.slash + Constants.taskProcessorIds +//
                        Constants.slash + taskProcessorInfo.getTaskProcessorId();//
        //可能其他地方把该taskProcessorId给取消了
        if (this.getZooKeeper().exists(taskProcessorIdPath, false) == null)
        {
            taskProcessorInfo.setRegister(false);//不存在的时候才设置为false
            return false;
        } else
        {
            Timestamp oldHeartBeatTimestamp = taskProcessorInfo.getHeartBeatTimestamp();
            taskProcessorInfo.setHeartBeatTimestamp(heartBeatTimestamp);
            taskProcessorInfo.setVersion(taskProcessorInfo.getVersion() + 1);//v+1
            String value = this.gson.toJson(taskProcessorInfo);
            try
            {
                this.getZooKeeper().setData(taskProcessorIdPath, value.getBytes(), -1);
                return true;
            } catch (Exception e)
            {
                taskProcessorInfo.setHeartBeatTimestamp(oldHeartBeatTimestamp);
                taskProcessorInfo.setVersion(taskProcessorInfo.getVersion() - 1);//为什么-1,这样用来恢复原来的记录
                throw e;
            }

        }
    }

    public void reportTaskProcessorDealInfo(TaskProcessorInfo taskProcessorInfo, Statistics statistics) throws Exception
    {
        String taskProcessorIdPath = //
                this.tasksRootPath + //
                        Constants.slash + taskProcessorInfo.getTaskId()//
                        + Constants.slash + Constants.taskProcessorIds + //
                        Constants.slash + taskProcessorInfo.getTaskProcessorId();
        if (this.getZooKeeper().exists(taskProcessorIdPath, false) == null)
        {
            //
        } else
        {
            taskProcessorInfo.setDealInfo(statistics.getStatus());
            String value = this.gson.toJson(taskProcessorInfo);
            try
            {
                this.getZooKeeper().setData(taskProcessorIdPath, value.getBytes(), -1);
            } catch (Exception e)
            {
                e.printStackTrace();
                log.error(e.getMessage());
            }

        }
    }

    @Override
    public int clearExpireTaskProcessor(String taskId, long expireTime) throws Exception
    {
        int count = 0;

        String taskProcessorIdsPath = this.tasksRootPath + Constants.slash + taskId + Constants.slash + Constants.taskProcessorIds;
        if (this.getZooKeeper().exists(taskProcessorIdsPath, false) == null)
        {
            String taskIdPath = this.tasksRootPath + Constants.slash + taskId;// ../
            if (this.getZooKeeper().exists(taskIdPath, false) == null)
            {
                this.getZooKeeper().create(taskIdPath, null, this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
            }
            //
            this.getZooKeeper().create(taskProcessorIdsPath, null, this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
        }
        for (String taskProcessorId : this.getZooKeeper().getChildren(taskProcessorIdsPath, false))
        {
            try
            {
                Stat stat = this.getZooKeeper().exists(taskProcessorIdsPath + Constants.slash + taskProcessorId, false);
                if (stat != null && (zooKeeperCurrentTimeMillis() - stat.getMtime() > expireTime))
                {//删除过期的server
                    ZooKeepers.directoryStructureDelete(this.getZooKeeper(), taskProcessorIdsPath + Constants.slash + taskProcessorId);
                    count++;
                }
            } catch (Exception e)
            {

                count++;
            }
        }
        return count;
    }

    @Override
    public List<TaskProcessorInfo> getTaskProcessorInfoList(String taskId)
            throws Exception
    {
        List<TaskProcessorInfo> taskProcessorInfoList = new ArrayList<TaskProcessorInfo>();

        String taskProcessorIdsPath = this.tasksRootPath + Constants.slash + taskId + Constants.slash + Constants.taskProcessorIds;
        if (this.getZooKeeper().exists(taskProcessorIdsPath, false) == null)
        {
            return taskProcessorInfoList;
        }
        List<String> taskProcessorIdList = this.getZooKeeper().getChildren(taskProcessorIdsPath, false);
        Collections.sort(taskProcessorIdList, new Comparator<String>()
        {
            public int compare(String leftString, String rightString)
            {
                return leftString.substring(leftString.lastIndexOf(Constants.$) + 1).compareTo(
                        rightString.substring(rightString.lastIndexOf(Constants.$) + 1));
            }
        });
        for (String taskProcessorId : taskProcessorIdList)
        {
            try
            {
                String taskProcessorInfoBytes = new String(this.getZooKeeper().getData(taskProcessorIdsPath + Constants.slash + taskProcessorId, false, null));
                TaskProcessorInfo taskProcessorInfo = this.gson.fromJson(taskProcessorInfoBytes, TaskProcessorInfo.class);
                taskProcessorInfo.setLeaderTaskProcessorTimestamp(new Timestamp(this.zooKeeperCurrentTimeMillis()));
                taskProcessorInfoList.add(taskProcessorInfo);
            } catch (Exception e)
            {
                log.error(e.getMessage(), e);
            }
        }
        return taskProcessorInfoList;
    }

    public List<String> getTaskProcessorIdList(String taskId) throws Exception
    {

        String taskProcessorIdsPath = this.tasksRootPath + Constants.slash + taskId + Constants.slash + Constants.taskProcessorIds;
        if (this.getZooKeeper().exists(taskProcessorIdsPath, false) == null)
        {
            return new ArrayList<String>();
        }
        List<String> taskProcessorIdList = this.getZooKeeper().getChildren(taskProcessorIdsPath, false);
        Collections.sort(taskProcessorIdList, new Comparator<String>()
        {
            public int compare(String leftString, String rightString)
            {
                return leftString.substring(leftString.lastIndexOf("$") + 1).compareTo(
                        rightString.substring(rightString.lastIndexOf("$") + 1));
            }
        });
        return taskProcessorIdList;
    }

    @Override
    public List<TaskProcessorInfo> getTaskProcessorInfoList(
            String taskId,//
            String ip, //
            String orderString) throws Exception
    {//


        List<TaskProcessorInfo> taskProcessorInfoList = new ArrayList<TaskProcessorInfo>();
        List<TaskProcessorInfo> temporaryTaskProcessorInfoList = this.getTaskProcessorInfoList(taskId);
        if (ip == null)
        {
            taskProcessorInfoList.addAll(temporaryTaskProcessorInfoList);
        } else
        {
            for (TaskProcessorInfo taskProcessorInfo : temporaryTaskProcessorInfoList)
            {
                //ip限制
                if (ip.equals(taskProcessorInfo.getIp()))
                {
                    taskProcessorInfoList.add(taskProcessorInfo);
                }
            }
        }
        Collections.sort(taskProcessorInfoList, new TaskProcessorComparator(orderString));//进行比较

        return taskProcessorInfoList;
    }

    public List<TaskProcessorInfo> getTaskProcessorInfoListByJvmInstanceId(String jvmInstanceId)
            throws Exception
    {
        List<TaskProcessorInfo> taskProcessorInfoList = new ArrayList<TaskProcessorInfo>();
        for (String taskId : this.getZooKeeper().getChildren(this.tasksRootPath, false))
        {
            String taskProcessorIdsPath = this.tasksRootPath + Constants.slash + taskId + Constants.slash + Constants.taskProcessorIds;
            for (String taskProcessorId : this.getZooKeeper().getChildren(taskProcessorIdsPath, false))
            {
                String taskProcessorInfoString = new String(this.getZooKeeper().getData(taskProcessorIdsPath + Constants.slash + taskProcessorId, false, null));
                TaskProcessorInfo taskProcessorInfo = this.gson.fromJson(taskProcessorInfoString, TaskProcessorInfo.class);
                taskProcessorInfo.setLeaderTaskProcessorTimestamp(new Timestamp(this.zooKeeperCurrentTimeMillis()));
                if (taskProcessorInfo.getJvmInstanceId().equals(jvmInstanceId))//全部遍历 找到machineId相关的task Processor
                {
                    taskProcessorInfoList.add(taskProcessorInfo);
                }
            }
        }
        Collections.sort(taskProcessorInfoList, new Comparator<TaskProcessorInfo>()
        {
            public int compare(TaskProcessorInfo leftTaskProcessorInfo, TaskProcessorInfo rightTaskProcessorInfo)
            {
                int result = leftTaskProcessorInfo.getTaskId().compareTo(rightTaskProcessorInfo.getTaskId());
                if (result == 0)
                {
                    String leftTaskProcessorId = leftTaskProcessorInfo.getTaskProcessorId();
                    String rightTaskProcessorId = rightTaskProcessorInfo.getTaskProcessorId();
                    result = leftTaskProcessorId.substring(leftTaskProcessorId.lastIndexOf(Constants.$) + 1).compareTo(
                            rightTaskProcessorId.substring(rightTaskProcessorId.lastIndexOf(Constants.$) + 1));
                }
                return result;
            }
        });
        return taskProcessorInfoList;
    }

    public boolean isLeaderTaskProcessorId(List<String> taskProcessorIdList, String taskProcessorId)
    {
        return taskProcessorId.equals(getMinTaskProcessorId(taskProcessorIdList));
    }

    /**
     * 当前最小的任务处理器是否是分配任务的处理器,如果是则表示系统处于稳定的状态
     */
    public boolean minTaskProcessorIdIsLeaderTaskProcessorId(String taskId) throws Exception
    {
        String minTaskProcessorId = this.getMinTaskProcessorId(this.getTaskProcessorIdList(taskId));//min id
        String taskItemsPath = this.tasksRootPath + Constants.slash + taskId + Constants.slash + Constants.taskItems;
        if (this.getZooKeeper().exists(taskItemsPath, false) != null)
        {
            byte[] taskItemsDataWithTaskProcessorIdBytes = this.getZooKeeper().getData(taskItemsPath, false, null);
            if (taskItemsDataWithTaskProcessorIdBytes != null && new String(taskItemsDataWithTaskProcessorIdBytes).equals(minTaskProcessorId))
            {
                return true;
            }
        }
        return false;
    }

    public String getMinTaskProcessorId(List<String> taskProcessorIdList)
    {
        if (taskProcessorIdList == null || taskProcessorIdList.size() == 0)
        {
            return null;
        }
        long num = Long.MAX_VALUE;
        long $num = -1;
        String leaderTaskProcessorId = null;
        for (String taskProcessorId : taskProcessorIdList)
        {
            $num = Long.parseLong(taskProcessorId.substring(taskProcessorId.lastIndexOf(Constants.$) + 1));//serverName $之后为index
            if (num > $num)
            {//min
                num = $num;
                leaderTaskProcessorId = taskProcessorId;
            }
        }
        return leaderTaskProcessorId;
    }

    public long getTaskProcessors$Stat$Version(String taskId) throws Exception
    {

        String taskProcessorIdsPath = this.tasksRootPath + Constants.slash + taskId + Constants.slash + Constants.taskProcessorIds;
        Stat stat = new Stat();
        this.getZooKeeper().getData(taskProcessorIdsPath, false, stat);
        return stat.getVersion();
    }

    public Map<String, Stat> getTaskProcessorIdToStatMap(String taskId) throws Exception
    {
        Map<String, Stat> taskProcessorIdToStatMap = new HashMap<String, Stat>();

        String taskProcessorIdsPath = this.tasksRootPath + Constants.slash + taskId + Constants.slash + Constants.taskProcessorIds;
        List<String> taskProcessorIdList = this.getZooKeeper().getChildren(taskProcessorIdsPath, false);
        for (String taskProcessorId : taskProcessorIdList)
        {
            String taskProcessorIdPath = taskProcessorIdsPath + Constants.slash + taskProcessorId;
            Stat stat = this.getZooKeeper().exists(taskProcessorIdPath, false);
            taskProcessorIdToStatMap.put(taskProcessorId, stat);
        }
        return taskProcessorIdToStatMap;
    }

    /**
     * <pre>
     * ========================================================================================================================
     * TaskItem
     * ========================================================================================================================
     * </pre>
     */
    public void createTaskItems(String taskId, String[] taskItemStringAndParameterStrings) throws Exception
    {
        TaskItem[] taskItems = new TaskItem[taskItemStringAndParameterStrings.length];
        Pattern pattern = Pattern.compile("\\s*:\\s*\\{");

        for (int i = 0; i < taskItemStringAndParameterStrings.length; i++)
        {
            taskItems[i] = new TaskItem();

            taskItems[i].setTaskId(taskId);
            //进行任务项index+任务项参数的分离
            Matcher matcher = pattern.matcher(taskItemStringAndParameterStrings[i]);
            if (matcher.find())
            {
                taskItems[i].setTaskItemString(taskItemStringAndParameterStrings[i].substring(0, matcher.start()).trim());
                taskItems[i].setParameter(taskItemStringAndParameterStrings[i].substring(matcher.end(), taskItemStringAndParameterStrings[i].length() - 1).trim());
            } else
            {
                taskItems[i].setTaskItemString(taskItemStringAndParameterStrings[i]);
            }
        }
        createTaskItemsWithTaskItemStrings(taskItems);
    }

    public void createTaskItemsWithTaskItemStrings(TaskItem[] taskItems) throws Exception
    {
        for (TaskItem taskItem : taskItems)
        {
            String taskItemsPath = this.tasksRootPath + Constants.slash + taskItem.getTaskId() + Constants.slash + Constants.taskItems;
            if (this.getZooKeeper().exists(taskItemsPath, false) == null)
            {
                ZooKeepers.createPath(this.getZooKeeper(), taskItemsPath, this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
            }
            String taskItemPath = taskItemsPath + Constants.slash + taskItem.getTaskItemString();
            this.getZooKeeper().create(taskItemPath, null, this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
            this.getZooKeeper().create(taskItemPath + Constants.$current_task_processor_id, null, this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
            this.getZooKeeper().create(taskItemPath + Constants.$request_task_processor_id, null, this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
            this.getZooKeeper().create(taskItemPath + Constants.$parameter, taskItem.getParameter().getBytes(), this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
            this.getZooKeeper().create(taskItemPath + Constants.$deal_info, taskItem.getDealInfo().getBytes(), this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
        }
    }

    public void releaseOldTaskItemsAndReAllotTaskItemInTaskProcessorIdList(String taskId, String taskProcessorId)
            throws Exception
    {


        String taskItemsPath = this.tasksRootPath + Constants.slash + taskId + Constants.slash + Constants.taskItems;
        try
        {
            ZooKeepers.directoryStructureDelete(this.getZooKeeper(), taskItemsPath);
        } catch (Exception e)
        {
            if (e instanceof KeeperException
                    && ((KeeperException) e).code().intValue() == KeeperException.Code.SESSIONEXPIRED.intValue())
            {
                log.warn("delete : zookeeper sessionzookeeper");
                zooKeeperManager.reConnect();
                ZooKeepers.directoryStructureDelete(this.getZooKeeper(), taskItemsPath);
            }
        }
        this.getZooKeeper().create(taskItemsPath, null, this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
        this.createTaskItems(taskId, this.getTaskUserDefinition(taskId).getTaskItemStrings());
        this.setTaskItemsLeaderTaskProcessorId(taskId, taskProcessorId);
    }


    @Override
    public int clearTaskItemCurrentTaskProcessorIdNodeWhenNotFoundInTaskProcessorIdList(String taskId,//
                                                                                        List<String> taskProcessorIdList) throws Exception
    {

        String taskItemsPath = this.tasksRootPath + Constants.slash + taskId + Constants.slash + Constants.taskItems;

        int count = 0;
        for (String taskItem : this.getZooKeeper().getChildren(taskItemsPath, false))
        {
            byte[] currentTaskProcessorIdBytes = this.getZooKeeper().getData(taskItemsPath + Constants.slash + taskItem + Constants.$current_task_processor_id, false, null);
            if (currentTaskProcessorIdBytes != null)
            {
                String currentTaskProcessorId = new String(currentTaskProcessorIdBytes);
                boolean isFind = false;
                for (String taskProcessorId : taskProcessorIdList)
                {
                    if (currentTaskProcessorId.equals(taskProcessorId))
                    {
                        isFind = true;
                        break;
                    }
                }
                if (!isFind)
                {//非法的处理器都要被清理
                    this.getZooKeeper().setData(taskItemsPath + Constants.slash + taskItem + Constants.$current_task_processor_id, null, -1);
                    count = count + 1;
                }
            } else
            {
                count = count + 1;
            }
        }
        return count;
    }

    public void setTaskItemsLeaderTaskProcessorId(String taskId, String taskProcessorId) throws Exception
    {
        String taskItemsPath = this.tasksRootPath + Constants.slash + taskId + Constants.slash + Constants.taskItems;
        this.getZooKeeper().setData(taskItemsPath, taskProcessorId.getBytes(), -1);
    }

    public void updateTaskItem(String taskId, String taskItem, String message) throws Exception
    {
        String taskItemPath = this.tasksRootPath + Constants.slash + taskId + Constants.slash + Constants.taskItems + Constants.slash + taskItem;

        //
        if (this.getZooKeeper().exists(taskItemPath + Constants.$deal_info, false) == null)
        {
            if (message == null)
            {
                message = "";
            }
            this.getZooKeeper().setData(taskItemPath + Constants.$deal_info, message.getBytes(), -1);
        }
    }

    /**
     * @param leaderTaskProcessorId 进行任务分配的taskProcessorId
     */
    @Override
    public void reallocateTaskItemsTaskCurrentTaskProcessorId(String taskId,//
                                                              List<String> taskProcessorIdList, //
                                                              String leaderTaskProcessorId
    ) throws Exception
    {
        //前面设置稳定标志的taskProcessorId,如果当前taskProcessorId为控制taskProcessorId，则可以进行任务分配
        if (this.isLeaderTaskProcessorId(taskProcessorIdList, leaderTaskProcessorId) == false)
        {
            return;
        }
        if (taskProcessorIdList.size() <= 0)
        {
            return;
        }

        String taskItemsPath = this.tasksRootPath + Constants.slash + taskId + Constants.slash + Constants.taskItems;
        List<String> taskItemList = this.getZooKeeper().getChildren(taskItemsPath, false);

        Collections.sort(taskItemList, new Comparator<String>()
        {
            public int compare(String leftString, String rightString)
            {
                if (StringUtils.isNumeric(leftString) && StringUtils.isNumeric(rightString))
                {
                    int leftIntValue = Integer.parseInt(leftString);
                    int rightIntValue = Integer.parseInt(rightString);
                    if (leftIntValue == rightIntValue)
                    {
                        return 0;
                    } else if (leftIntValue > rightIntValue)
                    {
                        return 1;
                    } else
                    {
                        return -1;
                    }
                } else
                {
                    return leftString.compareTo(rightString);
                }
            }
        });
        //

        int[] nums = Allots.allot(taskItemList.size(), taskProcessorIdList.size());//taskDataItemList size/taskProcessorId num
        //
        int accumulativeCount = 0;
        int taskProcessorIdIndex = 0;
        //
        int unModifyCount = 0;
        //为什么不按照处理器进行设置呢,因为按照分配算法计算出来的可能有的处理器没有任务项.
        for (int i = 0; i < taskItemList.size(); i++)
        {//可能出现有的处理器上面一个任务也分配不到的情况,任务项小于处理器数量

            if (taskProcessorIdIndex < taskProcessorIdList.size() //
                    && //
                    //i是否满足当下的区段
                    i >= (accumulativeCount + nums[taskProcessorIdIndex]))//index from 0,so index++
            {
                //如果不满足则进入下一区段
                accumulativeCount = accumulativeCount + nums[taskProcessorIdIndex];
                taskProcessorIdIndex = taskProcessorIdIndex + 1;
            }

            //
            String emptyString = "";
            String taskProcessorId = emptyString;
            //理论上都应该有值
            if (taskProcessorIdIndex < taskProcessorIdList.size())
            {
                taskProcessorId = taskProcessorIdList.get(taskProcessorIdIndex);//在最新的处理器上进行分配task item
            }
            //
            String taskItem = taskItemList.get(i);
            byte[] currentTaskProcessorIdBytes = this.getZooKeeper().getData(taskItemsPath + Constants.slash + taskItem + Constants.$current_task_processor_id, false, null);
            byte[] requestTaskProcessorIdBytes = this.getZooKeeper().getData(taskItemsPath + Constants.slash + taskItem + Constants.$request_task_processor_id, false, null);
            //currentTaskProcessorId  requestTaskProcessorId 在当前可能的取值是null(N),taskProcessorId(A),!taskProcessorId(B)这三种情况.
            //整体一共则9种情况
            /**
             * currentTaskProcessorId优先处理
             * <pre>
             *     currentTaskProcessorId  requestTaskProcessorId
             *     N                       N                    应该是currentTaskProcessorId NotAllowContinueDealWithTaskDataItemException,requestTaskProcessorId不变
             *     N                       A                    应该是currentTaskProcessorId NotAllowContinueDealWithTaskDataItemException,requestTaskProcessorId不变
             *     N                       B                    应该是currentTaskProcessorId NotAllowContinueDealWithTaskDataItemException,requestTaskProcessorId不变
             *
             * =============================================================================================================
             *     A                       N                    应该是currentTaskProcessorId 不变,requestTaskProcessorId不变
             *
             *     这两项需要覆盖原有申请的,保证原有申请的处理器最活跃(例外处理)
             *     A                       A                    应该是currentTaskProcessorId 不变,requestTaskProcessorId不变
             *     A                       B                    应该是currentTaskProcessorId 不变,requestTaskProcessorId不变
             * =============================================================================================================
             *     这三项都尝试申请
             *     B                       N                    应该是currentTaskProcessorId 不变,requestTaskProcessorId NotAllowContinueDealWithTaskDataItemException
             *     B                       A                    应该是currentTaskProcessorId 不变,requestTaskProcessorId不变[补X] 可以尝试申请
             *     B                       B                    应该是currentTaskProcessorId 不变,requestTaskProcessorId NotAllowContinueDealWithTaskDataItemException
             * </pre>
             */


            //
            if (currentTaskProcessorIdBytes == null || new String(currentTaskProcessorIdBytes).equals(emptyString))//"" 只是针对当个节点进行申请.
            {//还没有被分配
                this.getZooKeeper().setData(taskItemsPath + Constants.slash + taskItem + Constants.$current_task_processor_id, taskProcessorId.getBytes(), -1);
                this.getZooKeeper().setData(taskItemsPath + Constants.slash + taskItem + Constants.$request_task_processor_id, null, -1);
            } else if (currentTaskProcessorIdBytes != null && new String(currentTaskProcessorIdBytes).equals(taskProcessorId) && requestTaskProcessorIdBytes == null)
            {//已经占用,且currentTaskProcessorId=taskProcessorId
                unModifyCount = unModifyCount + 1;
            } else
            {//可以尝试申请(可能会覆盖其他值)
                this.getZooKeeper().setData(taskItemsPath + Constants.slash + taskItem + Constants.$request_task_processor_id, taskProcessorId.getBytes(), -1);
            }
        }
        //
        if (unModifyCount < taskItemList.size())
        {
            this.setTaskNeedReload(taskId);
        }

    }

    /**
     * 释放当前taskProcessorId,提升为请求taskProcessorId
     */
    @Override
    public void releaseTaskItemCurrentTaskProcessorId(String taskId, String taskProcessorId) throws Exception
    {

        String taskItemsPath = this.tasksRootPath + Constants.slash + taskId + Constants.slash + Constants.taskItems;
        boolean isModify = false;
        for (String taskItemString : this.getZooKeeper().getChildren(taskItemsPath, false))
        {
            byte[] currentTaskProcessorIdBytes = this.getZooKeeper().getData(taskItemsPath + Constants.slash + taskItemString + Constants.$current_task_processor_id, false, null);
            byte[] requestTaskProcessorIdBytes = this.getZooKeeper().getData(taskItemsPath + Constants.slash + taskItemString + Constants.$request_task_processor_id, false, null);

            if (requestTaskProcessorIdBytes != null && currentTaskProcessorIdBytes != null &&//
                    taskProcessorId.equals(new String(currentTaskProcessorIdBytes)))//释放当前taskProcessorId,提升为请求taskProcessorId
            {
                this.getZooKeeper().setData(taskItemsPath + Constants.slash + taskItemString + Constants.$current_task_processor_id, requestTaskProcessorIdBytes, -1);
                this.getZooKeeper().setData(taskItemsPath + Constants.slash + taskItemString + Constants.$request_task_processor_id, null, -1);
                isModify = true;
                break;
            }
        }
        if (isModify == true)
        {
            this.setTaskNeedReload(taskId);
        }
    }


    public void deleteTaskItem(String taskId, String taskItem) throws Exception
    {
        String taskItemPath = this.tasksRootPath + Constants.slash + taskId + Constants.slash + taskId + Constants.slash + Constants.taskItems + Constants.slash + taskItem;
        ZooKeepers.directoryStructureDelete(this.getZooKeeper(), taskItemPath);
    }

    public List<TaskItem> getTaskItemList(String taskId) throws Exception
    {
        List<TaskItem> taskItemList = new ArrayList<TaskItem>();
        String taskItemsPath = this.tasksRootPath + Constants.slash + taskId + Constants.slash + Constants.taskItems;
        if (this.getZooKeeper().exists(taskItemsPath, false) == null)
        {
            return taskItemList;
        }
        List<String> taskItemStringList = this.getZooKeeper().getChildren(taskItemsPath, false);
        Collections.sort(taskItemStringList, new Comparator<String>()
        {
            public int compare(String leftString, String rightString)
            {
                if (StringUtils.isNumeric(leftString) && StringUtils.isNumeric(rightString))
                {
                    int leftIntString = Integer.parseInt(leftString);
                    int rightIntString = Integer.parseInt(rightString);
                    if (leftIntString == rightIntString)
                    {
                        return 0;
                    } else if (leftIntString > rightIntString)
                    {
                        return 1;
                    } else
                    {
                        return -1;
                    }
                } else
                {
                    return leftString.compareTo(rightString);
                }
            }
        });// end
        for (String taskItemString : taskItemStringList)
        {
            TaskItem taskItem = new TaskItem();
            taskItem.setTaskId(taskId);
            taskItem.setTaskItemString(taskItemString);
            String taskItemPath = taskItemsPath + Constants.slash + taskItemString;
            //
            byte[] currentServerIdBytes = this.getZooKeeper().getData(taskItemPath + Constants.$current_task_processor_id, false, null);
            if (currentServerIdBytes != null)
            {
                taskItem.setCurrentServerId(new String(currentServerIdBytes));
            }
            //
            byte[] requestServerIdBytes = this.getZooKeeper().getData(taskItemPath + Constants.$request_task_processor_id, false, null);
            if (requestServerIdBytes != null)
            {
                taskItem.setRequestServerId(new String(requestServerIdBytes));
            }
            //
            byte[] parameterBytes = this.getZooKeeper().getData(taskItemPath + Constants.$parameter, false, null);
            if (parameterBytes != null)
            {
                taskItem.setParameter(new String(parameterBytes));
            }
            //
            byte[] dealInfoBytes = this.getZooKeeper().getData(taskItemPath + Constants.$deal_info, false, null);
            if (dealInfoBytes != null)
            {
                taskItem.setDealInfo(new String(dealInfoBytes));
            }
            taskItemList.add(taskItem);
        }
        return taskItemList;

    }

    @Override
    public int getTaskItemListSize(String taskId) throws Exception
    {

        String taskItemsPath = this.tasksRootPath + Constants.slash + taskId + Constants.slash + Constants.taskItems;
        return this.getZooKeeper().getChildren(taskItemsPath, false).size();
    }

    /**
     * 获取任务项
     */
    @Override
    public List<TaskItemStringAndParameterString> getTaskItemStringAndParameterStringListWithTaskProcessorId(String taskId, String taskProcessorId)
            throws Exception
    {

        String taskItemsPath = this.tasksRootPath + Constants.slash + taskId + Constants.slash + Constants.taskItems;
        //
        List<String> taskItemStringList = this.getZooKeeper().getChildren(taskItemsPath, false);
        Collections.sort(taskItemStringList, new Comparator<String>()
        {
            public int compare(String leftString, String rightString)
            {
                if (StringUtils.isNumeric(leftString) && StringUtils.isNumeric(rightString))
                {
                    int leftIntValue = Integer.parseInt(leftString);
                    int rightIntValue = Integer.parseInt(rightString);
                    if (leftIntValue == rightIntValue)
                    {
                        return 0;
                    } else if (leftIntValue > rightIntValue)
                    {
                        return 1;
                    } else
                    {
                        return -1;
                    }
                } else
                {
                    return leftString.compareTo(rightString);
                }
            }
        });//end


        List<TaskItemStringAndParameterString> taskItemStringAndParameterStringList = new ArrayList<TaskItemStringAndParameterString>();
        for (String taskItemString : taskItemStringList)//何处进行分配
        {
            byte[] currentTaskProcessorIdBytes = this.getZooKeeper().getData(taskItemsPath + Constants.slash + taskItemString + Constants.$current_task_processor_id, false, null);
            //taskProcessorId.equals(currentServerId)
            if (currentTaskProcessorIdBytes != null && taskProcessorId.equals(new String(currentTaskProcessorIdBytes)))
            {
                TaskItemStringAndParameterString taskItemStringAndParameterString = new TaskItemStringAndParameterString();
                taskItemStringAndParameterString.setTaskItemString(taskItemString);//必须存在任务项,可能没有参数 0:{x1=x1,x2=x2},1:{x1=x1,x2=x2}   0,1这两种模式
                byte[] parameterBytes = this.getZooKeeper().getData(taskItemsPath + Constants.slash + taskItemString + Constants.$parameter, false, null);
                if (parameterBytes != null)
                {
                    taskItemStringAndParameterString.setParameterString(new String(parameterBytes));
                }
                taskItemStringAndParameterStringList.add(taskItemStringAndParameterString);


            }
        }
        return taskItemStringAndParameterStringList;
    }


}



 